package com.wudl.flink.stream.tablesql


import org.apache.flink._
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.sources.CsvTableSource

object CreateTableFile {
  def main(args: Array[String]): Unit = {

    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val envtSettings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, envtSettings)


    //读取数据
    val tableSource = new CsvTableSource("/station.log", Array[String]("f1", "f2", "f3", "f4", "f5", "f6"), Array(Types.STRING, Types.STRING, Types.STRING, Types.STRING, Types.LONG, Types.LONG))
    // 注册一张表
    tableEnv.registerTableSource("t_station_log", tableSource)
    val table: Table = tableEnv.scan("t_station_log")
    table.printSchema()
    table.dropColumns("f6")
    println("---------------------------")

  }

}
